package org.csdn.titan

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.csdn.titan.utils.SparkUtils

/*
 * @Date:2021/2/19-22:39
 * @Author:fcc
 * @NickName:江南云朵
 * @QQ:46210953
 * @Description:XXXXXXXXXXX
 */
object LogMapID2 {

  def main(args: Array[String]): Unit = {
    val spark:SparkSession = SparkUtils.getSparkSession(this.getClass.getName)

    val applog:Dataset[String] = spark.read.textFile("data/yiee_logs/2020-01-12/app")
    val weblog:Dataset[String] = spark.read.textFile("data/yiee_logs/2020-01-12/web")
    val wxapplog:Dataset[String] = spark.read.textFile("data/yiee_logs/2020-01-12/wxapp")



    val appIDS = extractIDs(applog)
    val webIDS = extractIDs(weblog)
    val wxIDS = extractIDs(wxapplog)

    val all_ids:RDD[Array[String]] = appIDS.union(webIDS).union(wxIDS)


    //构造点集合vertices
    val vertices:RDD[(Long,String)] = all_ids.flatMap(groups => {
      for (temp_id <- groups) yield (temp_id.hashCode.toLong, temp_id)
    })

    //构造边集合edges
    val edges:RDD[Edge[String]] = all_ids.flatMap(groups => {
      for (i <- 0 to groups.length - 2) yield Edge(groups(i).hashCode.toLong, groups(i + 1).hashCode.toLong, "")
    }).map(edge => (edge, 1))
      .reduceByKey(_ + _)
      .filter(tuple => tuple._2 > 2)
      .map(tuple => tuple._1)

    //将上一日的映射字典加载进来，变成点、边集合
    import spark.implicits._
    val preMapIds:DataFrame = spark.read.parquet("data/yiee_logs/2020-01-11/guidmap")
    val preVertices:RDD[(VertexId,String)] = preMapIds.rdd.map(row => {
      val id_hashcode = row.getAs[VertexId]("id_hashcode")
      val guid = row.getAs[VertexId]("guid")
      (id_hashcode, "")
//      case Row(idFlag: VertexId, guid: VertexId) => (idFlag, "")


    })

   import spark.implicits._
    val preEdges = preMapIds.rdd.map(row => {
      val id_hashcode = row.getAs[VertexId]("id_hashcode")
      val guid = row.getAs[VertexId]("guid")
      Edge(id_hashcode, guid,"")
    })


    //构建图
    val graph = Graph(vertices.union(preVertices), edges.union(preEdges))
    //得到连通的点集
    val connectGraph = graph.connectedComponents()
    val result_vertices:VertexRDD[VertexId] = connectGraph.vertices


    val preIDsMap = preMapIds.rdd.map(row => {
      val id_hashcode = row.getAs[VertexId]("id_hashcode")
      val guid = row.getAs[VertexId]("guid")
      (id_hashcode, guid)
    }).collectAsMap()
    val v_bc = spark.sparkContext.broadcast(preIDsMap)

    //将结果与上一日的对比，调整guid
    val last_result_vertices = result_vertices
      .map(tp => (tp._2, tp._1))
      .groupByKey()
      .mapPartitions(iter => {
        iter.map(tp => {
          val idsMap = v_bc.value
          var todayGuid = tp._1
          val todayIds = tp._2
          var find = false
          for (tempid <- todayIds if !find) {
            val maybeGuid: Option[VertexId] = idsMap.get(tempid)
            if (maybeGuid.isDefined) {
              todayGuid = maybeGuid.get
              find = true
            }

          }
          (todayGuid, todayIds)

        })

      })
      .flatMap(tp => {
        val guid = tp._1
        val todayIds = tp._2
        for (id <- todayIds) yield (id, guid)
      })
//      .map(tp => {
//        val idsMap = v_bc.value
//        var todayGuid = tp._1
//        val todayIds = tp._2
//        var find = false
//        for(tempid <- todayIds if !find) {
//          val maybeGuid:Option[VertexId] = idsMap.get(tempid)
//          if (maybeGuid.isDefined) {
//            todayGuid = maybeGuid.get
//            find = true
//          }
//
//        }
//        (todayGuid,todayIds)
//
//      })

    //保存结果
    import spark.implicits._
    last_result_vertices.toDF("id_hashcode","guid").write.parquet("data/yiee_logs/2020-01-12/guidmap")
    spark.close()
  }



  def extractIDs(logs:Dataset[String]):RDD[Array[String]] = {

    logs.rdd.map(line =>{
      val jsonObject = JSON.parseObject(line)
      val userObject = jsonObject.getJSONObject("user");
      val uid = userObject.getString("uid")
      val phoneObj = userObject.getJSONObject("phone")
      val imei = phoneObj.getString("imei")
      val mac = phoneObj.getString("mac")
      val imsi = phoneObj.getString("imsi")
      val androidId = phoneObj.getString("androidId")
      val deviceId = phoneObj.getString("deviceId")
      val uuid = phoneObj.getString("uuid")
      Array(uid,imei,mac,imsi,androidId,deviceId,uuid).filter(StringUtils.isNotBlank(_))
    })

  }

}
